package com.shujia.core

import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.runtime.operators.util.{AssignerWithPeriodicWatermarksAdapter, AssignerWithPunctuatedWatermarksAdapter}

import java.time.Duration

object Demo05EventTime {
  def main(args: Array[String]): Unit = {

    /**
     * 统计每个用户出现的次数
     */

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    /**
     * 基于数据本身的时间触发计算任务
     * 1、需要将时间属性设置为 TimeCharacteristic.EventTime
     * 2、向Flink程序注册时间字段
     *
     * 水位线：默认等于当前接收到的数据中最大的时间戳
     * 解决数据的乱序到达问题，可以将水位线前移
     *
     * 触发点：例如每5s中触发一次任务，那么一天就会有 24*3600/5 = 17280个触发点
     * 第一个触发点在：0时0分5秒
     * 第二个触发点在：0时0分10秒
     * 第三个触发点在：0时0分15秒
     * ......
     * 最后一个触发点：24时0分0秒
     *
     * 任务的触发条件：
     * 1、窗口中有数据
     * 2、水位线超过当前窗口的触发点
     */
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    env.setParallelism(1)

    // 无界流
    val linesKVDS: DataStream[(String, Long)] =
      env
        .socketTextStream("master", 8888)
        .map(line => {
          val arr: Array[String] = line.split(",")
          (arr(0), arr(1).toLong)
        })

    // 指定将数据中的第二列作为时间（事件时间）
    //    val assignTimeLinesDS: DataStream[String] = linesDS.assignAscendingTimestamps(line => line.split(",")(1).toLong)

    // 将水位线前移5s并设置事件时间
    val assignTimeAndWatermarkDS: DataStream[(String, Long)] = linesKVDS
      .assignTimestampsAndWatermarks(
        WatermarkStrategy
          // 控制将水位线前移多少 前移5s
          .forBoundedOutOfOrderness[(String, Long)](Duration.ofSeconds(5))
          // 指定将哪一列作为 事件时间
          .withTimestampAssigner(new SerializableTimestampAssigner[(String, Long)] {
            override def extractTimestamp(element: (String, Long), recordTimestamp: Long): Long = {
              //将第二列作为时间戳（事件时间）
              element._2
            }
          }))

//    val assignTimeAndWatermarkDS: DataStream[(String, Long)] = linesKVDS
//      .assignTimestampsAndWatermarks(
//        // 将水位线前移5s
//        new BoundedOutOfOrdernessTimestampExtractor[(String, Long)](Time.seconds(5)) {
//          // 提取出数据中的时间 作为 事件时间
//          override def extractTimestamp(element: (String, Long)): Long = element._2
//        })


    assignTimeAndWatermarkDS
      .map(kv => (kv._1, 1))
      .keyBy(_._1)
      .timeWindow(Time.seconds(5))
      .sum(1)
      .print()

    env.execute()


  }

}
